package org.ala.tiktools.rpc.kcp.netty.server;

import io.netty.buffer.ByteBuf;
import io.netty.channel.*;
import io.netty.channel.nio.AbstractNioChannel;
import io.netty.channel.nio.AbstractNioMessageChannel;
import io.netty.util.internal.PlatformDependent;
import io.netty.util.internal.SocketUtils;
import io.netty.util.internal.logging.InternalLogger;
import io.netty.util.internal.logging.InternalLoggerFactory;
import org.ala.tiktools.rpc.kcp.netty.KcpPacket;
import org.ala.tiktools.rpc.kcp.netty.server.config.DefaultKcpServerChannelConfig;
import org.ala.tiktools.rpc.kcp.netty.IKcpServerChannelConfig;
import org.ala.tiktools.rpc.kcp.netty.server.handler.KcpServerChannelCloser;
import org.ala.tiktools.rpc.kcp.netty.server.handler.KcpServerChannelCreator;
import org.ala.tiktools.rpc.kcp.netty.server.handler.KcpServerChannelReader;
import org.ala.tiktools.rpc.kcp.netty.server.handler.KcpServerChannelWriter;
import org.ala.tiktools.rpc.kcp.protocol.IKcpOutput;
import org.ala.tiktools.rpc.kcp.protocol.Kcp;

import java.io.IOException;
import java.net.SocketAddress;
import java.net.SocketException;
import java.nio.channels.DatagramChannel;
import java.nio.channels.SelectionKey;
import java.nio.channels.spi.SelectorProvider;
import java.util.*;


/**
 * server端channel
 *
 * @author ala
 * @date 2025-04-14 13:00
 */
public class KcpServerChannel extends AbstractNioMessageChannel implements ServerChannel {
    public static final InternalLogger log = InternalLoggerFactory.getInstance(KcpServerChannel.class);


    /**
     *  子channel（虚拟channel，底层并不真实绑定某个socket套接字。udp协议也没长连socket给你绑）
     */
    protected Map<SocketAddress, KcpServerChildChannel> childChannelMap = new HashMap<>();
    /**
     *  channel 配置
     */
    protected IKcpServerChannelConfig config;
    /**
     *  nio 轮训器
     */
    private static final SelectorProvider DEFAULT_SELECTOR_PROVIDER = SelectorProvider.provider();
    /**
     * 写入工具
     */
    protected IKcpOutput output = new KcpServerOutput();
    /**
     *  附加信息
     */
    private static final ChannelMetadata METADATA = new ChannelMetadata(false);
    /**
     *  Bootstrap给的worker线程组
     */
    protected EventLoopGroup worker;
    /**
     *  本serverChannel监听的端口号
     */
    protected int port;


    /**
     *  各种读取工具
     */
    protected KcpServerChannelCloser closer;
    protected KcpServerChannelCreator creator;
    protected KcpServerChannelReader reader;
    protected KcpServerChannelWriter writer;


    /**
     *  上次update时间戳
     */
    private int tsUpdate;
    /**
     *  定时update
     */
    private boolean scheduleUpdate;


    public KcpServerChannel() {
        this(newSocket(DEFAULT_SELECTOR_PROVIDER));
    }
    public KcpServerChannel(SelectorProvider provider) {
        this(newSocket(provider));
    }
    public KcpServerChannel(DatagramChannel socket) {
        super(null, socket, SelectionKey.OP_READ);
        config = new DefaultKcpServerChannelConfig(this, socket.socket());
        closer = new KcpServerChannelCloser(this);
        creator = new KcpServerChannelCreator(this);
        reader = new KcpServerChannelReader(this);
        writer = new KcpServerChannelWriter(this);
    }
    private static DatagramChannel newSocket(SelectorProvider provider) {
        try {
            return provider.openDatagramChannel();
        } catch (IOException e) {
            throw new RuntimeException("无法打开一个udp channel");
        }
    }


    @Override
    public IKcpServerChannelConfig config() {
        return config;
    }
    @Override
    public boolean isActive() {
        DatagramChannel ch = javaChannel();
        return ch.isOpen() && ch.socket().isBound();
    }
    @Override
    public DatagramChannel javaChannel() {
        return (DatagramChannel) super.javaChannel();
    }
    @Override
    public ChannelMetadata metadata() { return METADATA; }


    /**
     *  接收数据
     */
    @Override
    protected int doReadMessages(List<Object> datas) throws Exception {
        return reader.doReadMessages(datas);
    }
    @Override
    public boolean closeOnReadError(Throwable cause) {
        if (cause instanceof SocketException) {
            return false;
        }
        return super.closeOnReadError(cause);
    }


    /**
     *  将数据写给javaChannel
     */
    @Override
    protected boolean doWriteMessage(Object msg, ChannelOutboundBuffer in) throws Exception {
        return writer.doWriteMessage(msg, in);
    }
    /**
     *  消息过滤
     */
    @Override
    protected Object filterOutboundMessage(Object msg) throws Exception {
        if (msg instanceof KcpPacket) {
            KcpPacket p = (KcpPacket) msg;
            ByteBuf content = p.getContent();
            if (isSingleDirectBuffer(content)) {
                return p;
            }
            content.retain();   // 引用计数+1，因为 newDirectBuffer 里会给content的引用计数-1，这里先加回来
            KcpPacket np = KcpPacket.newInstance(newDirectBuffer(content), p.getRemoteAddress());
            p.release();
            return np;
        }
        throw new UnsupportedOperationException("只接受KcpPacket类型参数");
    }
    /**
     * 检测ByteBuf是否只有1个引用
     */
    private static boolean isSingleDirectBuffer(ByteBuf buf) {
        return buf.isDirect() && buf.nioBufferCount() == 1;
    }
    @Override
    protected boolean continueOnWriteError() {
        return true;
    }


    /**
     * 初始化绑定
     */
    @Override
    protected void doBind(SocketAddress localAddress) throws Exception {
        //  抄 NioServerSocketChannel
        if (PlatformDependent.javaVersion() >= 7) {
            SocketUtils.bind(javaChannel(), localAddress);
        } else {
            javaChannel().socket().bind(localAddress);
        }
    }
    /**
     * 关闭
     * <p>  关闭ServerChannel下记录的所有KcoClientChannel
     * <p>  清理每个KcoClientChannel绑定的资源
     */
    @Override
    protected void doClose() throws Exception {
        closer.doClose();;
    }
    /**
     *  关闭childChannel
     */
    protected void doCloseChildChannel(KcpServerChildChannel childChannel) {
        closer.doCloseChildChannel(childChannel);
    }


    /**
     *  连接相关
     */
    @Override
    protected boolean doConnect(SocketAddress remoteAddress, SocketAddress localAddress) throws Exception {
        throw new UnsupportedOperationException();
    }
    @Override
    protected void doFinishConnect() throws Exception {
        throw new UnsupportedOperationException();
    }
    @Override
    protected void doDisconnect() throws Exception {
        throw new UnsupportedOperationException();
    }


    /**
     *  update子channel
     */
    public void updateChildKcp(KcpServerChildChannel channel) {
        writer.updateChildKcp(channel);
    }


    /**
     * 远程地址相关
     */
    @Override
    protected SocketAddress localAddress0() { return javaChannel().socket().getLocalSocketAddress(); }
    @Override
    protected SocketAddress remoteAddress0() { return javaChannel().socket().getRemoteSocketAddress(); }
    @Override
    public SocketAddress localAddress() { return javaChannel().socket().getLocalSocketAddress(); }
    @Override
    public SocketAddress remoteAddress() { return javaChannel().socket().getRemoteSocketAddress(); }


    /**
     *  Unsafe
     */
    @Override
    protected KcpServeUnsafe newUnsafe() {
        return new KcpServeUnsafe();
    }
    private final class KcpServeUnsafe extends AbstractNioUnsafe {
        private final List<Object> readBuf = new ArrayList<Object>();
        @Override
        public void read() {
            reader.unsafeReadInBoss(readBuf, recvBufAllocHandle());
        }
    }

    /**
     *  写实现
     */
    class KcpServerOutput implements IKcpOutput {
        @Override
        public void out(ByteBuf buf, Kcp kcp) {
            KcpServerChildChannel channel = (KcpServerChildChannel) kcp.getChannel();
            AbstractNioChannel.NioUnsafe unsafe = unsafe();
            unsafe.write(KcpPacket.newInstance(buf, channel.remoteAddress()), unsafe.voidPromise());
            unsafe.flush();
        }
    }


    public Map<SocketAddress, KcpServerChildChannel> getChildChannelMap() {
        return childChannelMap;
    }

    public void setChildChannelMap(Map<SocketAddress, KcpServerChildChannel> childChannelMap) {
        this.childChannelMap = childChannelMap;
    }

    public IKcpServerChannelConfig getConfig() {
        return config;
    }

    public void setConfig(IKcpServerChannelConfig config) {
        this.config = config;
    }

    public IKcpOutput getOutput() {
        return output;
    }

    public void setOutput(IKcpOutput output) {
        this.output = output;
    }

    public int getTsUpdate() {
        return tsUpdate;
    }

    public void setTsUpdate(int tsUpdate) {
        this.tsUpdate = tsUpdate;
    }

    public boolean isScheduleUpdate() {
        return scheduleUpdate;
    }

    public void setScheduleUpdate(boolean scheduleUpdate) {
        this.scheduleUpdate = scheduleUpdate;
    }

    public KcpServerChannelCloser getCloser() {
        return closer;
    }

    public void setCloser(KcpServerChannelCloser closer) {
        this.closer = closer;
    }

    public KcpServerChannelCreator getCreator() {
        return creator;
    }

    public void setCreator(KcpServerChannelCreator creator) {
        this.creator = creator;
    }

    public KcpServerChannelReader getReader() {
        return reader;
    }

    public void setReader(KcpServerChannelReader reader) {
        this.reader = reader;
    }

    public KcpServerChannelWriter getWriter() {
        return writer;
    }

    public void setWriter(KcpServerChannelWriter writer) {
        this.writer = writer;
    }

    public EventLoopGroup getWorker() { return worker; }

    public void setWorker(EventLoopGroup worker) { this.worker = worker; }

    public int getPort() { return port; }

    public void setPort(int port) { this.port = port; }
}
